package com.xxd.dfs.backupnode;


import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.CharBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.nio.charset.Charset;
import java.nio.charset.CharsetDecoder;
import java.util.Iterator;

/**
 * @author: XiaoDong.Xie
 * @create: 2020-08-27 11:14
 * @description:
 */
public class TransferFsImage {
    private static final Logger logger = LoggerFactory.getLogger(TransferFsImage.class);
    private static CharsetDecoder decoder = Charset.forName("UTF-8").newDecoder();

    /**
     * 把fsimage 传输给namenode
     *
     * @param fsimage
     */
    public static void sendFSImage2NameNode(FSImage fsimage) {
        // 建立Socket 连接
        SocketChannel channel = null;
        Selector selector = null;
        try {
            channel = SocketChannel.open();
            channel.configureBlocking(false);
            channel.connect(new InetSocketAddress(9000));

            selector = Selector.open();
            channel.register(selector, SelectionKey.OP_CONNECT);

            boolean uploading = true;
            while (uploading) {
                selector.select();

                Iterator<SelectionKey> keyIterator = selector.selectedKeys().iterator();
                while (keyIterator.hasNext()) {
                    SelectionKey key = keyIterator.next();
                    keyIterator.remove();

                    if (key.isConnectable()) { // 连接请求

                        channel = (SocketChannel) key.channel();

                        if (channel.isConnectionPending()) { // 正在连接
                            channel.finishConnect(); // 等待完成连接

                            ByteBuffer buffer = ByteBuffer.wrap(fsimage.getFsimageJson().getBytes());
                            logger.info("client 连接完毕，准备发送数据, 数据大小为 = {}", buffer.capacity());
                            channel.write(buffer);

                            key.interestOps(SelectionKey.OP_READ);
                        }
                    } else if (key.isReadable()) { // 读取数据的请求
                        channel = (SocketChannel) key.channel();
                        ByteBuffer buffer = ByteBuffer.allocate(1024);
                        int count = channel.read(buffer);
                        if (count > 0) {
                            buffer.flip();

                            String response = new String(buffer.array(), 0, count);

                            logger.info("上传fsimage文件成功，响应消息为：{}", response);
                            channel.close();
                            uploading = false;
                        }
                    }
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (channel != null) {
                try {
                    channel.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }

            if (selector != null) {
                try {
                    selector.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }


}
